package com.atguigu.Flink.sql.query;

import com.atguigu.Flink.POJO.WindSpeedSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;
import static org.apache.flink.table.api.Expressions.$;

public class Flink04_GroupingWindowApi {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WindSpeedSensor> ds = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new WindSpeedSensor(fields[0].trim(), Integer.valueOf(fields[1].trim()), Long.valueOf(fields[2].trim()));
                        }

                );
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Schema schema = Schema.newBuilder()
                .column("id", "STRING")
                .column("vc" , "INT")
                .column("ts" , "BIGINT")
                .columnByExpression("pt" , "proctime()")
                .columnByExpression("et" , "TO_TIMESTAMP_LTZ(ts, 3) ")
                .watermark("et" ,  "et - INTERVAL '0' SECOND ")
                .build();
        Table table = tableEnv.fromDataStream(ds, schema);

        // 滚动窗口
        // 计数滚动
        TumbleWithSizeOnTimeWithAlias w1 = Tumble.over(rowInterval(2L)).on($("pt")).as("w");
        // 时间滚动
        TumbleWithSizeOnTimeWithAlias w2 = Tumble.over(lit(5).seconds()).on($("pt")).as("w");
        TumbleWithSizeOnTimeWithAlias w3 = Tumble.over(lit(5).seconds()).on($("et")).as("w");

        //滑动窗口
        //计数滑动
        //计数滑动，窗口的首次计算必须要等收集到窗口大小相同的数据。
        SlideWithSizeAndSlideOnTimeWithAlias w4 = Slide.over(rowInterval(6L)).every(rowInterval(3L)).on($("pt")).as("w");
        //时间滑动
        SlideWithSizeAndSlideOnTimeWithAlias w5 = Slide.over(lit(10).seconds()).every(lit(5).seconds()).on($("pt")).as("w");
        SlideWithSizeAndSlideOnTimeWithAlias w6 = Slide.over(lit(10).seconds()).every(lit(5).seconds()).on($("et")).as("w");

        //会话窗口
        SessionWithGapOnTimeWithAlias w7 = Session.withGap(lit(3).seconds()).on($("pt")).as("w");
        SessionWithGapOnTimeWithAlias w8 = Session.withGap(lit(3).seconds()).on($("et")).as("w");


        //使用窗口
        table.window(w8)
                .groupBy($("w")  , $("id"))
                .select( /*$("id") , $("vc") ,*/ $("w").start(), $("w").end() ,   $("id")  , $("vc").sum())
                .execute()
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
